package com.clw.template.storage;

import com.aliyun.oss.OSSClient;
import com.aliyun.oss.internal.OSSHeaders;
import com.aliyun.oss.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * @author Yogeek
 * @date 2018/7/16 16:10
 * @decrpt 阿里云对象存储服务
 */
public class AliyunStorage implements Storage {
    public static final Logger LOGGER = LoggerFactory.getLogger(AliyunStorage.class);
    private  String endpoint;
    private  String accessKeyId;
    private  String accessKeySecret;
    private  String bucketName;


    public String getEndpoint() {
        return endpoint;
    }

    public void setEndpoint(String endpoint) {
        this.endpoint = endpoint;
    }

    public String getAccessKeyId() {
        return accessKeyId;
    }

    public void setAccessKeyId(String accessKeyId) {
        this.accessKeyId = accessKeyId;
    }

    public String getAccessKeySecret() {
        return accessKeySecret;
    }

    public void setAccessKeySecret(String accessKeySecret) {
        this.accessKeySecret = accessKeySecret;
    }

    public String getBucketName() {
        return bucketName;
    }

    public void setBucketName(String bucketName) {
        this.bucketName = bucketName;
    }

    /**
     * 获取阿里云OSS客户端对象
     *
     * @return ossClient
     */
    private OSSClient getOSSClient(){
        return new OSSClient(endpoint,accessKeyId, accessKeySecret);
    }

    private String getBaseUrl() {
        return "https://" + bucketName + "." +  endpoint + "/" ;
    }
    /**
     * 阿里云OSS对象存储简单上传实现
     */
    @Override
    public void store(InputStream inputStream, long contentLength, String contentType, String keyName) {
        try {
            final int size = 1 * 1024 * 1024;
            if (contentLength > size){
                multipartUpdate(inputStream,contentLength,contentType,keyName);
            } else {
                streamingUpdate(inputStream,contentLength,contentType,keyName);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

    }

    public void streamingUpdate(InputStream inputStream, long contentLength, String contentType, String keyName) throws IOException{
        // 简单文件上传, 最大支持 5 GB, 适用于小文件上传, 建议 20M以下的文件使用该接口
        ObjectMetadata objectMetadata = new ObjectMetadata();
        objectMetadata.setContentLength(contentLength);
        objectMetadata.setContentType(contentType);
        // 对象键（Key）是对象在存储桶中的唯一标识。
        PutObjectRequest putObjectRequest = new PutObjectRequest(bucketName, keyName, inputStream, objectMetadata);
        PutObjectResult putObjectResult = getOSSClient().putObject(putObjectRequest);
    }

    private static ByteArrayOutputStream cloneInputStream(InputStream input) {
        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            byte[] buffer = new byte[1024];
            int len;
            while ((len = input.read(buffer)) > -1) {
                baos.write(buffer, 0, len);
            }
            baos.flush();
            return baos;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public Stream<Path> loadAll() {
        return null;
    }

    @Override
    public Path load(String keyName) {
        return null;
    }

    @Override
    public Resource loadAsResource(String keyName) {
        try {
            URL url = new URL(getBaseUrl() + keyName);
            Resource resource = new UrlResource(url);
            if (resource.exists() || resource.isReadable()) {
                return resource;
            } else {
                return null;
            }
        } catch (MalformedURLException e) {
            e.printStackTrace();
            return null;
        }
    }

    @Override
    public void delete(String keyName) {
        try {
            getOSSClient().deleteObject(bucketName, keyName);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    @Override
    public String generateUrl(String keyName) {
        return getBaseUrl() + keyName;
    }

    /**
     * 多线程分片上传
     * @param inputStream
     * @param contentLength
     * @param contentType
     * @param keyName
     * @throws IOException
     */
    public void multipartUpdate(InputStream inputStream, long fileLength, String contentType, String keyName) throws IOException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        List<PartETag> partETags = Collections.synchronizedList(new ArrayList<PartETag>());
        // 创建InitiateMultipartUploadRequest对象。
        InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, keyName);

        // 如果需要在初始化分片时设置文件存储类型，请参考以下示例代码。
        ObjectMetadata metadata = new ObjectMetadata();
        metadata.setHeader(OSSHeaders.OSS_STORAGE_CLASS, StorageClass.Standard.toString());
        request.setObjectMetadata(metadata);

        // 初始化分片。
        InitiateMultipartUploadResult upresult = getOSSClient().initiateMultipartUpload(request);
        // 返回uploadId，它是分片上传事件的唯一标识，您可以根据这个ID来发起相关的操作，如取消分片上传、查询分片上传等。
        String uploadId = upresult.getUploadId();

        // 计算文件有多少个分片。
        final long partSize = 1 * 1024 * 1024L;   // 1MB
        int partCount = (int) (fileLength / partSize);
        if (fileLength % partSize != 0) {
            partCount++;
        }
        if (partCount > 10000) {
            throw new RuntimeException("Total parts count should not exceed 10000");
        } else {
            LOGGER.info("Total parts count " + partCount + "\n");
        }
        ByteArrayOutputStream baos = cloneInputStream(inputStream);

        // 遍历分片上传。
        for (int i = 0; i < partCount; i++) {
            long startPos = i * partSize;
            long curPartSize = (i + 1 == partCount) ? (fileLength - startPos) : partSize;
            executorService.execute(new PartUploader(new ByteArrayInputStream(baos.toByteArray()), startPos, curPartSize, i + 1, uploadId, bucketName, keyName, getOSSClient(),partETags));
        }

        executorService.shutdown();
        while (!executorService.isTerminated()) {
            try {
                executorService.awaitTermination(5, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if (partETags.size() != partCount) {
            throw new IllegalStateException("Upload multiparts fail due to some parts are not finished yet");
        } else {
            LOGGER.info("Succeed to complete multiparts into an object named " + keyName + "");
        }

        listAllParts(uploadId,keyName);
        completeMultipartUpload(uploadId,keyName,partETags);

    }

    private void listAllParts(String uploadId, String key) {
        LOGGER.info("Listing all parts......");
        ListPartsRequest listPartsRequest = new ListPartsRequest(bucketName, key, uploadId);
        PartListing partListing = getOSSClient().listParts(listPartsRequest);

        int partCount = partListing.getParts().size();
        for (int i = 0; i < partCount; i++) {
            PartSummary partSummary = partListing.getParts().get(i);
            LOGGER.info("Part#" + partSummary.getPartNumber() + ", ETag=" + partSummary.getETag());
        }
    }

    private void completeMultipartUpload(String uploadId, String key, List<PartETag> partETags) {
        // Make part numbers in ascending order
        Collections.sort(partETags, new Comparator<PartETag>() {

            @Override
            public int compare(PartETag p1, PartETag p2) {
                return p1.getPartNumber() - p2.getPartNumber();
            }
        });

        LOGGER.info("Completing to upload multiparts");
        CompleteMultipartUploadRequest completeMultipartUploadRequest =
                new CompleteMultipartUploadRequest(bucketName, key, uploadId, partETags);
        getOSSClient().completeMultipartUpload(completeMultipartUploadRequest);
    }

    private static class PartUploader implements Runnable {

        private InputStream instream;
        private long startPos;

        private long partSize;
        private int partNumber;
        private String uploadId;
        private String bucketName;
        private String key;
        private OSSClient client;
        private List<PartETag> partETags;

        public PartUploader(InputStream instream, long startPos, long partSize, int partNumber, String uploadId, String bucketName, String key, OSSClient client, List<PartETag> partETags) {
            this.instream = instream;
            this.startPos = startPos;
            this.partSize = partSize;
            this.partNumber = partNumber;
            this.uploadId = uploadId;
            this.bucketName = bucketName;
            this.key = key;
            this.client = client;
            this.partETags = partETags;
        }

        @Override
        public void run() {
            try {
                instream.skip(this.startPos);

                UploadPartRequest uploadPartRequest = new UploadPartRequest();
                uploadPartRequest.setBucketName(bucketName);
                uploadPartRequest.setKey(key);
                uploadPartRequest.setUploadId(this.uploadId);
                uploadPartRequest.setInputStream(instream);
                uploadPartRequest.setPartSize(this.partSize);
                uploadPartRequest.setPartNumber(this.partNumber);

                UploadPartResult uploadPartResult = client.uploadPart(uploadPartRequest);
                LOGGER.info("Part#" + this.partNumber + " done");
                synchronized (partETags) {
                    partETags.add(uploadPartResult.getPartETag());
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (instream != null) {
                    try {
                        instream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}
